
#include "hsf.h"
#include "mqttlib.h"
#include "m2m_socket.h"
#include "socket_driver.h"


#define MQTT_SERVER_PORT  1883

#define MQTT_KEEPALIVE_SEC  (180)

socket_para g_mqtt_socket_para;

mqtt_broker_handle_t g_stMQTTBroker;

extern uint8_t mqtt_establish_ok;
extern void upgrade_start(char *url);
extern uint8_t heartbeat_count;


char reg_topic_name[96]={'0'};
char sub_topic_name[96]={'0'};
char pub_topic_name[96]={'0'};
char upgrade_topic_name[96]={'0'};
char heartbeat_topic_name[96]={'0'};


#if defined(WX_TOPIC)
	 char reg_topic_name_pre[32] = "/fw/alphair2/";
	 char sub_topic_name_pre[32] = "/fw/alphair2/";
	 char pub_topic_name_pre[32] = "/fw/alphair2/";
	 char upgrade_topic_name_pre[32] = "/fw/alphair2/";		
	 char heartbeat_topic_name_pre[32] = "/fw/";	
	 char sub_topic_name_end[32] = "/down";		 
	 char reg_topic_name_end[32] = "/server";
	 char pub_topic_name_end[32] = "/up";
	 char upgrade_topic_name_end[32] = "/upgrade";
     char heartbeat_topic_name_end[32] = "heartbeat";	 
	 #define MQTT_SERVER_ADDR  "iotplc.cn"
#elif defined(ORKAN_TOPIC)
	 char reg_topic_name_pre[32] = "/fw/orkan/";
	 char sub_topic_name_pre[32] = "/fw/orkan/";
	 char pub_topic_name_pre[32] = "/fw/orkan/";
	 char upgrade_topic_name_pre[32] = "/fw/orkan/";
	 char heartbeat_topic_name_pre[32] = "/fw/";			
	 char reg_topic_name_end[32] = "/v1.0/server";
	 char sub_topic_name_end[32] = "/v1.0/down";
	 char pub_topic_name_end[32] = "/v1.0/up";
	 char upgrade_topic_name_end[32] = "/v1.0/upgrade";	
     char heartbeat_topic_name_end[32] = "heartbeat";	
	 #define MQTT_SERVER_ADDR  "iotplc.cn"
#elif defined(ORKAN_TOPIC_XF_V1_1)
	 char reg_topic_name_pre[32] = "/fw/orkan/";
	 char sub_topic_name_pre[32] = "/fw/orkan/";
	 char pub_topic_name_pre[32] = "/fw/orkan/";
	 char upgrade_topic_name_pre[32] = "/fw/orkan/";
	 char heartbeat_topic_name_pre[32] = "/fw/";			
	 char reg_topic_name_end[32] = "/v1.1/server";
	 char sub_topic_name_end[32] = "/v1.1/down";
	 char pub_topic_name_end[32] = "/v1.1/up";
	 char upgrade_topic_name_end[32] = "/v1.1/upgrade";	
     char heartbeat_topic_name_end[32] = "heartbeat";
	 #define MQTT_SERVER_ADDR  "iotplc.cn"
#elif defined(JINGGE_TOPIC)
	 char reg_topic_name_pre[32] = "/fw/jingge/";
	 char sub_topic_name_pre[32] = "/fw/jingge/";
	 char pub_topic_name_pre[32] = "/fw/jingge/";
	 char upgrade_topic_name_pre[32] = "/fw/jingge/";
	 char heartbeat_topic_name_pre[32] = "/fw/";			
	 char reg_topic_name_end[32] = "/v1.0/server";
	 char sub_topic_name_end[32] = "/v1.0/down";
	 char pub_topic_name_end[32] = "/v1.0/up";
	 char upgrade_topic_name_end[32] = "/v1.0/upgrade";	
     char heartbeat_topic_name_end[32] = "heartbeat";
	#define MQTT_SERVER_ADDR  "mqtt.jingge.iotplc.cn"			  	 	
#elif defined(JIABANG_TOPIC)
	char reg_topic_name_pre[32] = "/fw/jiabang/";
	char sub_topic_name_pre[32] = "/fw/jiabang/";
	char pub_topic_name_pre[32] = "/fw/jiabang/";
	char upgrade_topic_name_pre[32] = "/fw/jiabang/";	
	char heartbeat_topic_name_pre[32] = "/fw/";			
	char reg_topic_name_end[32] = "/v1.0/server";
	char sub_topic_name_end[32] = "/v1.0/down";
	char pub_topic_name_end[32] = "/v1.0/up";
	char upgrade_topic_name_end[32] = "/v1.0/upgrade";	
	char heartbeat_topic_name_end[32] = "heartbeat";	 
	#define MQTT_SERVER_ADDR  "iotplc.cn"			
#elif defined(DIJIA_TOPIC)
	char reg_topic_name_pre[32] = "/fw/dijia/";
	char sub_topic_name_pre[32] = "/fw/dijia/";
	char pub_topic_name_pre[32] = "/fw/dijia/";
	char upgrade_topic_name_pre[32] = "/fw/dijia/";	
	char heartbeat_topic_name_pre[32] = "/fw/";					
	char reg_topic_name_end[32] = "/v1.0/server";
	char sub_topic_name_end[32] = "/v1.0/down";
	char pub_topic_name_end[32] = "/v1.0/up";
	char upgrade_topic_name_end[32] = "/v1.0/upgrade";	
	char heartbeat_topic_name_end[32] = "heartbeat";
	#define MQTT_SERVER_ADDR  "iotplc.cn"		
#elif defined(ZEFENG_TOPIC)
	char reg_topic_name_pre[32] = "/fw/zefeng/";
	char sub_topic_name_pre[32] = "/fw/zefeng/";
	char pub_topic_name_pre[32] = "/fw/zefeng/";
	char upgrade_topic_name_pre[32] = "/fw/zefeng/";
	char heartbeat_topic_name_pre[32] = "/fw/";				
	char reg_topic_name_end[32] = "/v1.0/server";
	char sub_topic_name_end[32] = "/v1.0/down";
	char pub_topic_name_end[32] = "/v1.0/up";
	char upgrade_topic_name_end[32] = "/v1.0/upgrade";
	char heartbeat_topic_name_end[32] = "heartbeat";
	#define MQTT_SERVER_ADDR  "iotplc.cn"
#elif defined(LJN_TOPIC)
	char reg_topic_name_pre[32] = "/fw/lvjianneng/";
	char sub_topic_name_pre[32] = "/fw/lvjianneng/";
	char pub_topic_name_pre[32] = "/fw/lvjianneng/";
	char upgrade_topic_name_pre[32] = "/fw/lvjianneng/";
	char heartbeat_topic_name_pre[32] = "/fw/";				
	char reg_topic_name_end[32] = "/v1.0/server";
	char sub_topic_name_end[32] = "/v1.0/down";
	char pub_topic_name_end[32] = "/v1.0/up";
	char upgrade_topic_name_end[32] = "/v1.0/upgrade";
	char heartbeat_topic_name_end[32] = "heartbeat";
	#define MQTT_SERVER_ADDR  "iotplc.cn"	
#elif defined(ZHONGQING_TOPIC)
	char reg_topic_name_pre[32] = "/fw/zhongqing/";
	char sub_topic_name_pre[32] = "/fw/zhongqing/";
	char pub_topic_name_pre[32] = "/fw/zhongqing/";
	char upgrade_topic_name_pre[32] = "/fw/zhongqing/";
	char heartbeat_topic_name_pre[32] = "/fw/";				
	char reg_topic_name_end[32] = "/v1.0/server";
	char sub_topic_name_end[32] = "/v1.0/down";
	char pub_topic_name_end[32] = "/v1.0/up";
	char upgrade_topic_name_end[32] = "/v1.0/upgrade";
	char heartbeat_topic_name_end[32] = "heartbeat";
	#define MQTT_SERVER_ADDR  "iotplc.cn"						 	
#endif


char client_id_pre[]="lpt120";
char client_id[32]={'0'};



topic_para *g_reg_topic = NULL;
topic_para *g_sub_topic = NULL;
topic_para *g_pub_topic= NULL;
topic_para *g_upgrade_topic= NULL;
topic_para *g_heartbeat_topic= NULL;

extern char get_mac[12];
extern uint8_t wifi_state_cloud[13];

PROCESS(mqtt_socket_process, "mqtt_socket_process");

void mqtt_socket_start(void)
{
	#if defined(_sys_debug_)
	   u_printf("mqtt_socket_start----------------------- \r\n");
    #endif
	process_start(&mqtt_socket_process, NULL);
}

void mqtt_socket_exit(void)
{
	process_exit(&mqtt_socket_process);
}


void mqtt_tcp_socket_init(void)
{
	g_mqtt_socket_para.fd = -1;
	g_mqtt_socket_para.connect_status = SOCKET_CLOSED;
	#if defined(_sys_debug_)
		u_printf("mqtt_tcp_socket_init \r\n");
	#endif
}

void mqtt_process_init_start(void)
{
	mqtt_tcp_socket_init();
	app_mqtt_init(&g_stMQTTBroker);
	mqtt_tcp_connect(MQTT_SERVER_ADDR, MQTT_SERVER_PORT);
}

void mqtt_tcp_connect(char *addr, int port)
{
	uip_ip4addr_t remote_ip_addr;
	if(uiplib_ipaddrconv(addr, &remote_ip_addr) == 0)
	{
		resolv_query(addr);
		return;
	}
	else
	{
		g_mqtt_socket_para.connect_status = SOCKET_CREATE;
		g_mqtt_socket_para.fd = tcpconnect(&remote_ip_addr, port, &mqtt_socket_process);
		#if defined(_sys_debug_)
			u_printf("Tcp socket(%d) connectting to [%d.%d.%d.%d:%d]\n", g_mqtt_socket_para.fd, 
			remote_ip_addr.u8[0], remote_ip_addr.u8[1], remote_ip_addr.u8[2], remote_ip_addr.u8[3], port);
	    #endif

	}
}


void app_mqtt_init(mqtt_broker_handle_t *broker)
{
    memset(client_id,0x00,sizeof(client_id));
	strcat(client_id,client_id_pre) ;
	strcat(client_id,get_mac) ;	
	mqtt_init(broker, client_id);
	mqtt_init_auth(broker, NULL, NULL);
	mqtt_set_alive(broker, MQTT_KEEPALIVE_SEC);
	#if defined(_sys_debug_)
		u_printf("client_id=======%s \r\n",client_id);
    #endif	
}

int mqtt_send_packet(void* socket_info, const void* buf, unsigned int count)
{
	NETSOCKET fd = *((NETSOCKET*)socket_info);
	int ret = 0;
	ret = tcpsend(fd, buf, count);
	if(ret < 0)
	{
		#if defined(_sys_debug_)
		  if(ret < 0)  u_printf("tcp send fail  ret===%d\n",ret);
		#endif
		hfsys_softreset();
	}
	return ret;
}


void mqtt_login_server(mqtt_broker_handle_t *broker)
{
	broker->socket_info = (void*)&(g_mqtt_socket_para.fd);
	broker->send = mqtt_send_packet;

	if(mqtt_connect(broker)== -1)
    {
		#if defined(_sys_debug_)
			u_printf("Mqtt login server fail!\n");
	    #endif	
	}
	else
	{
		#if defined(_sys_debug_)
			u_printf("Do mqtt login ...\n");
	    #endif
	}
	
}

void init_topic(topic_para **topic, char *name, int name_len)
{
	*topic = (topic_para *)hfmem_malloc(sizeof(topic_para) + name_len*sizeof(char));
	if(*topic == NULL)
	{
		#if defined(_sys_debug_)
			u_printf("NO memory!\n");
		#endif
		hfsys_softreset();
		return ;
	}

	memset(*topic,0, sizeof(topic_para) + name_len*sizeof(char));
	(*topic)->msg_id = 0;
	memcpy((*topic)->name, name, name_len);
}

void deinit_topic(topic_para **topic)
{
	hfmem_free(*topic);
	*topic = NULL;
}
extern uint8_t wifi_State;
uint16_t lp=0;
void app_parse_mqttmsg(uint8_t *packet_buffer)
{
	uint8_t msg_type = 0;
	uint16_t msg_id_rcv = 0;
	char topic_name[56]={0};
	char msg[128]={0};
	int i=0;		
	msg_type = MQTTParseMessageType(packet_buffer);
	//u_printf("app_parse_mqttmsg \r\n");	
	switch(msg_type)
	{
		case MQTT_MSG_CONNACK:
			#if defined(_sys_debug_)
				u_printf("packet_buffer[3] == %d,Mqtt login server success\n",packet_buffer[3]);
			#endif		
			if(packet_buffer[3] == 0)
			{
				//hfuart_send(HFUART0,wifi_state_cloud,sizeof(wifi_state_cloud),100);
				wifi_State=2;
                //u_printf("mqtt CONN ok \r\n");
			    if(get_mac!=NULL)
				{
				   //设备启动发送mac字符串到服务器
				   memset(reg_topic_name,0x00,sizeof(reg_topic_name));
  				   strcat(reg_topic_name,reg_topic_name_pre) ;
   				   strcat(reg_topic_name,get_mac) ;  
				   strcat(reg_topic_name,reg_topic_name_end) ;
				   #if defined(_sys_debug_)
					  u_printf("g_reg_topic_name_strcat==============%s\r\n",reg_topic_name);
				   #endif					   
                   //设备发送到服务器的主题
				   memset(pub_topic_name,0x00,sizeof(pub_topic_name));
  				   strcat(pub_topic_name,pub_topic_name_pre) ;
   				   strcat(pub_topic_name,get_mac) ;  
				   strcat(pub_topic_name,pub_topic_name_end) ;  
				   #if defined(_sys_debug_)
					  u_printf("g_pub_topic_name_strcat==============%s\r\n",pub_topic_name);
				   #endif						  
                   //订阅消息的主题
				   memset(sub_topic_name,0x00,sizeof(sub_topic_name));
				   strcat(sub_topic_name,sub_topic_name_pre) ;		 
				   strcat(sub_topic_name,get_mac) ;
				   #if defined(_sys_debug_)
					  u_printf("g_sub_topic_name_end_strcat==============%s\r\n",sub_topic_name_end);
					  u_printf("g_sub_topic_name_strcat==============%s\r\n",sub_topic_name);
				   #endif	  
				   #if defined(WX_TOPIC)
				       strcat(sub_topic_name,"/down") ; 
				   #elif defined(ORKAN_TOPIC)||defined(LJN_TOPIC)||defined(JINGGE_TOPIC) ||defined(DIJIA_TOPIC) ||defined(ZEFENG_TOPIC)||defined(JIABANG_TOPIC)||defined(ZHONGQING_TOPIC)	
				       strcat(sub_topic_name,"/v1.0/down") ; 
				   #elif defined(ORKAN_TOPIC_XF_V1_1)
				       strcat(sub_topic_name,"/v1.1/down") ; 
				   #endif	
   				   
				   #if defined(_sys_debug_)
					  u_printf("g_sub_topic_name_strcat==============%s\r\n",sub_topic_name);
				   #endif	
				   memset(upgrade_topic_name,0x00,sizeof(upgrade_topic_name));
  				   strcat(upgrade_topic_name,upgrade_topic_name_pre) ;
   				   strcat(upgrade_topic_name,get_mac) ;  
				   strcat(upgrade_topic_name,upgrade_topic_name_end) ;
				   #if defined(_sys_debug_)
				      u_printf("g_upgrade_topic_name_strcat==============%s\r\n",upgrade_topic_name);
				   #endif	
				   memset(heartbeat_topic_name,0x00,sizeof(heartbeat_topic_name));
				   strcat(heartbeat_topic_name,heartbeat_topic_name_pre) ;
				   //strcat(heartbeat_topic_name,get_mac) ;  
				   strcat(heartbeat_topic_name,heartbeat_topic_name_end) ; 	
				   #if defined(_sys_debug_)
				      u_printf("g_heartbeat_topic_name_strcat==============%s\r\n",heartbeat_topic_name);
				   #endif				   					  				   		   				      
				}
				else 
				{
                   //u_printf("get_mac is null\r\n");
				}
				
				deinit_topic(&g_reg_topic);
				deinit_topic(&g_pub_topic);		
				deinit_topic(&g_sub_topic);		
				deinit_topic(&g_upgrade_topic);		
				deinit_topic(&g_heartbeat_topic);						
				
				/*注册主题*/
				init_topic(&g_reg_topic, reg_topic_name, sizeof(reg_topic_name));
				/*发布主题*/
				init_topic(&g_pub_topic, pub_topic_name, sizeof(pub_topic_name));
				//u_printf("g_pub_topic topic[%s] success\n", g_pub_topic->name);				
				/*订阅主题*/
				init_topic(&g_sub_topic, sub_topic_name, sizeof(sub_topic_name));	
				#if defined(_sys_debug_)
				   u_printf("g_sub_topic_name_start==============%s\r\n",g_sub_topic->name);
			    #endif				
				/*升级主题*/
				init_topic(&g_upgrade_topic, upgrade_topic_name, sizeof(upgrade_topic_name));
				/*heartbeat主题*/
				init_topic(&g_heartbeat_topic, heartbeat_topic_name, sizeof(heartbeat_topic_name));				
	
				mqtt_subscribe(&g_stMQTTBroker, g_sub_topic->name, &(g_sub_topic->msg_id));
				mqtt_subscribe(&g_stMQTTBroker, g_upgrade_topic->name, &(g_upgrade_topic->msg_id));	
				mqtt_subscribe(&g_stMQTTBroker, g_heartbeat_topic->name, &(g_heartbeat_topic->msg_id));				
	
				mqtt_publish(&g_stMQTTBroker, g_reg_topic->name, get_mac,12, 0);
				
				hfuart_send(HFUART0,wifi_state_cloud,sizeof(wifi_state_cloud),100);

				mqtt_establish_ok=1;				
			}
			else
			{
				#if defined(_sys_debug_)
                	u_printf("login server failed  packet_buffer[3] ==== %d\r\n",packet_buffer[3]); 
				#endif					
			}
				
			break;

		case MQTT_MSG_SUBACK:
		
			msg_id_rcv = mqtt_parse_msg_id(packet_buffer);
			if(g_sub_topic && msg_id_rcv == g_sub_topic->msg_id) 
			{
				#if defined(_sys_debug_)
					u_printf("Subcribe topic[%s] success\n", g_sub_topic->name);
			    #endif					
			}
			break;

		case MQTT_MSG_PUBLISH:
            
			mqtt_parse_pub_topic(packet_buffer, topic_name);	
			
			uint16_t msg_len=mqtt_parse_publish_msg(packet_buffer, msg);
 			
            if(strcmp(sub_topic_name,topic_name)==0)
			{
				hfuart_send(HFUART0,msg,msg_len,100);
				#if defined(_sys_debug_)
					u_printf("recv_topic_msg===");	
					for(lp=0;lp<msg_len;lp++)  u_printf("%02x ",msg[lp]); 
					u_printf("\r\n");	
				#endif			
			}
			else if(strcmp(heartbeat_topic_name,topic_name)==0)
			{
				#if defined(_sys_debug_)
				  u_printf("recv heartbeat-----------------------------------------\r\n");
				#endif	
				heartbeat_count=0;		   
			}
			else if(strcmp(topic_name,upgrade_topic_name)==0)
			{
				if(strcmp("upgrade_fw",msg)==0)
				{
				   #if defined(WX_TOPIC) 
                   		upgrade_start("http://adm.iotplc.cn/company/fengwen/alphair/v1.0/lpt120-main_upgrade.bin");       
				   #elif defined(ORKAN_TOPIC)			
						upgrade_start("http://adm.iotplc.cn/company/orkan/v1.0/lpt120-main_upgrade.bin");
				   #elif defined(ORKAN_TOPIC_XF_V1_1)			
						upgrade_start("http://adm.iotplc.cn/company/orkan/v1.1/lpt120-main_upgrade.bin");						
				   #elif defined(DIJIA_TOPIC)				
						upgrade_start("http://adm.iotplc.cn/company/dijia/v1.0/lpt120-main_upgrade.bin");	
				   #elif defined(ZEFENG_TOPIC)				
						upgrade_start("http://adm.iotplc.cn/company/zefeng/v1.0/lpt120-main_upgrade.bin");
				   #elif defined(JIABANG_TOPIC)	
				        upgrade_start("http://adm.iotplc.cn/company/jiabang/v1.0/lpt120-main_upgrade.bin");			
				   #elif defined(ZHONGQING_TOPIC)
						upgrade_start("http://adm.iotplc.cn/company/zhongqing/v1.0/lpt120-main_upgrade.bin");
				   #elif defined(LJN_TOPIC)						
				        upgrade_start("http://adm.iotplc.cn/company/ljn/v1.0/lpt120-main_upgrade.bin");		
				   #elif defined(JINGGE_TOPIC)						
				        upgrade_start("http://adm.iotplc.cn/company/jingge/v1.0/lpt120-main_upgrade.bin");																							   					   
				   #endif
				   u_printf("upgrade start\r\n");
				}
				else if(strcmp("upgrade_fw_test",msg)==0)
				{
				   #if defined(WX_TOPIC)
                   		upgrade_start("http://adm.iotplc.cn/company/fengwen/alphair/test/lpt120-main_upgrade.bin");       
				   #elif defined(ORKAN_TOPIC) 			
						upgrade_start("http://adm.iotplc.cn/company/orkan/test/lpt120-main_upgrade.bin");
				   #elif defined(ORKAN_TOPIC_XF_V1_1)			
						upgrade_start("http://adm.iotplc.cn/company/orkan/test/lpt120-main_upgrade.bin");							
				   #elif defined(DIJIA_TOPIC)	
						upgrade_start("http://adm.iotplc.cn/company/dijia/test/lpt120-main_upgrade.bin");
				   #elif defined(ZEFENG_TOPIC)	
						upgrade_start("http://adm.iotplc.cn/company/zefeng/test/lpt120-main_upgrade.bin");	
				   #elif defined(JIABANG_TOPIC)	
						upgrade_start("http://adm.iotplc.cn/company/jiabang/test/lpt120-main_upgrade.bin");	
					#elif defined(ZHONGQING_TOPIC)
						upgrade_start("http://adm.iotplc.cn/company/zhongqing/test/lpt120-main_upgrade.bin");	
					#elif defined(LJN_TOPIC)						
				        upgrade_start("http://adm.iotplc.cn/company/ljn/test/lpt120-main_upgrade.bin");		
					#elif defined(JINGGE_TOPIC)						
				        upgrade_start("http://adm.iotplc.cn/company/jingge/test/lpt120-main_upgrade.bin");																													
				   #endif
				   #if defined(_sys_debug_)
				   	u_printf("upgrade test start\r\n");
			       #endif					   
				}				
			}

			break;

		case MQTT_MSG_UNSUBACK:
			msg_id_rcv = mqtt_parse_msg_id(packet_buffer);
			break;	

		case MQTT_MSG_PUBACK://Qos1
			msg_id_rcv = mqtt_parse_msg_id(packet_buffer);
			break;

		case MQTT_MSG_PUBREC://Qos2
			msg_id_rcv = mqtt_parse_msg_id(packet_buffer);
			break;

		case MQTT_MSG_PUBCOMP://Qos2
			msg_id_rcv = mqtt_parse_msg_id(packet_buffer);
			break;

		default:
			#if defined(_sys_debug_)
				u_printf("Unknow mqtt msg type\n");
			#endif			

			break;
	}
		
}

int tcp_set_send_flag=0;

PROCESS_THREAD(mqtt_socket_process, ev, data)
{
	PROCESS_BEGIN();
	
    SOCKETMSG msg;
	static struct etimer mqtt_socket_timer;
	int recv;

	#if defined(_sys_debug_)
		u_printf("PROCESS_THREAD(mqtt_socket_process, ev, data)----------------------- \r\n");
	#endif
	

	#if defined(_sys_debug_)
		u_printf("start_to_connect_server\r\n");
    #endif

	mqtt_tcp_socket_init();
	app_mqtt_init(&g_stMQTTBroker);
	mqtt_tcp_connect(MQTT_SERVER_ADDR, MQTT_SERVER_PORT);

	while(1) 
	{
		PROCESS_WAIT_EVENT();
		#if defined(_sys_debug_)
		   u_printf("PROCESS_THREAD---ev==============%d\r\n",ev);
        #endif        
		if(ev == resolv_event_found)
		{
			char *p_hostname = (char *)data;

			if(strcmp(p_hostname, MQTT_SERVER_ADDR))
				continue;
			
			uip_ipaddr_t addr;
			uip_ipaddr_t *p_addr = &addr;
			uip_ip4addr_t remote_ip_addr;
			resolv_lookup(p_hostname, &p_addr);
			uip_ipaddr_copy(&remote_ip_addr, p_addr);

			char ip_str[20]= {0}; 
			sprintf(ip_str,"%d.%d.%d.%d", remote_ip_addr.u8[0], remote_ip_addr.u8[1], remote_ip_addr.u8[2], remote_ip_addr.u8[3]);
		
			mqtt_tcp_connect(ip_str, MQTT_SERVER_PORT);	
		}
		else if(ev == PROCESS_EVENT_EXIT)
		{
			mqtt_disconnect(&g_stMQTTBroker);
			tcpclose(g_mqtt_socket_para.fd);
			#if defined(_sys_debug_)
				u_printf("PROCESS_EVENT_EXIT happen!\n");
	        #endif
			break;
		}
		else if(ev == PROCESS_EVENT_MSG)
		{
            msg = *(SOCKETMSG *)data;

			if(msg.status == SOCKET_CONNECTED) 
			{
				#if defined(_sys_debug_)
					u_printf("MQTT tcp socket(%d) connected\n", msg.socket);
			    #endif	
				g_mqtt_socket_para.connect_status = SOCKET_CONNECTED;
				tcp_set_send_flag=tcp_set_send_buffer_size(g_mqtt_socket_para.fd, MQTT_SOCKET_SEND_BUF_SIZE);
				#if defined(_sys_debug_)
					u_printf("tcp_set_send_buffer_size===%d\n", tcp_set_send_flag);
			    #endif					
				if(-1==tcp_set_send_flag)  hfsys_softreset();
				mqtt_login_server(&g_stMQTTBroker);
			}
			else if(msg.status == SOCKET_CLOSED)
			{
				if(g_mqtt_socket_para.connect_status == SOCKET_CREATE)
				{
					#if defined(_sys_debug_)
						u_printf("Socked(%d) create fail and retry...\n", msg.socket);
			    	#endif
				}
				else
				{
					#if defined(_sys_debug_)
						u_printf("MQTT tcp socket(%d) closed\n", msg.socket);
				    #endif
				}
				g_mqtt_socket_para.fd = -1;
				g_mqtt_socket_para.connect_status = SOCKET_CLOSED;
				etimer_set(&mqtt_socket_timer, 3*CLOCK_SECOND);
				PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&mqtt_socket_timer));
				mqtt_tcp_connect(MQTT_SERVER_ADDR, MQTT_SERVER_PORT);				
			}
			else if(msg.status == SOCKET_SENDACK)
			{
			
			}
			else if(msg.status == SOCKET_NEWDATA)
			{
				g_mqtt_socket_para.connect_status = SOCKET_NEWDATA;
				recv = tcprecv(g_mqtt_socket_para.fd, g_mqtt_socket_para.recv_buf, MQTT_SOCKET_RECV_BUF_SIZE);
				if(recv<=0)
				{
					#if defined(_sys_debug_)
						u_printf("tcprecv error! ret:%d\n", recv);
				    #endif
					continue;
				}
				app_parse_mqttmsg(g_mqtt_socket_para.recv_buf);	
			}
		}
	}
	PROCESS_END();
}








